package com.niodata.dt.fs;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.s3a.S3AFileSystem;

/**
 * Niodata dp s3a filesystem. s3a list empty dir,return dir itself,trim it.
 */
public class DpS3AFileSystem extends S3AFileSystem {


  @Override
  public FileStatus[] listStatus(Path f) throws
        IOException {
    Path orgPath = Path.getPathWithoutSchemeAndAuthority(f);
    FileStatus[] statuses = super.listStatus(orgPath);
    if (statuses == null || statuses.length == 0) {
      return statuses;
    }
    List<FileStatus> statusList = Arrays.stream(statuses).filter(
        x -> {
          Path path = Path.getPathWithoutSchemeAndAuthority(x.getPath());
          return !path.equals(orgPath);
        }
    ).collect(Collectors.toList());
    return statusList.toArray(new FileStatus[statusList.size()]);
  }
}
